package com.kqkj.dyrule.marketing.functions;

import com.kqkj.dyrule.marketing.beans.*;
import com.kqkj.dyrule.marketing.gauges.RuleIsMatchAvgTimeGauge;
import com.kqkj.dyrule.marketing.service.TriggerModelRuleMatchServiceImpl;
import com.kqkj.dyrule.marketing.utils.StateDescContainer;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.runtime.KieSession;

import java.util.Iterator;
import java.util.List;

public class RuleMatchKeyedProcessFunctionV2 extends KeyedBroadcastProcessFunction<String, DynamicKeyedBean, MaxwellLogBean, RuleMatchResult> {

    ListState<EventBean> listState;
    TriggerModelRuleMatchServiceImpl triggerModelRuleMatchService;
    ReadOnlyBroadcastState<String, List<RuleStateBean>> ruleBeanState;
    ListState<Tuple2<RuleStateBean, Long>> timeInfoState;
    Counter rule_calc_timeAmount;
    Counter rule_match_calc_timeAmount;
    Counter rule_not_match_calc_timeAmount;
    Counter rule_calc_cntAmount;
    Counter rule_is_match_calc_cntAmount;
    Counter rule_not_match_calc_cntAmount;
    RuleIsMatchAvgTimeGauge ruleIsMatchAvgTimeGauge;
    @Override
    public void open(Configuration parameters) throws Exception {
        listState = getRuntimeContext().getListState(StateDescContainer.getEventBeansDesc());
        triggerModelRuleMatchService = new TriggerModelRuleMatchServiceImpl(listState);
        //记录定时信息的state
        timeInfoState = getRuntimeContext().getListState(StateDescContainer.ruleTimerStateDesc);



        rule_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_calc_timeAmount");
        rule_match_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_is_match_calc_timeAmount");
        rule_not_match_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_not_match_calc_timeAmount");

        rule_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_calc_cntAmount");
        rule_is_match_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_is_match_calc_cntAmount");
        rule_not_match_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_not_match_calc_cntAmount");

        ruleIsMatchAvgTimeGauge = getRuntimeContext().getMetricGroup().gauge("计算成立的规则的平均每次计算耗时",new RuleIsMatchAvgTimeGauge());

    }

    @Override
    public void processElement(DynamicKeyedBean dynamicKeyedBean, ReadOnlyContext ctx, Collector<RuleMatchResult> out) throws Exception {
        if (ruleBeanState == null) ruleBeanState = ctx.getBroadcastState(StateDescContainer.ruleStateDesc);

        //遍历每一个规则
        List<RuleStateBean> ruleStateBeans = ruleBeanState.get(dynamicKeyedBean.getKeyNames());
        for (RuleStateBean ruleStateBean : ruleStateBeans) {
            MarketingRule marketingRule = ruleStateBean.getMarketingRule();
            KieSession kieSession = ruleStateBean.getKieSession();
            RuleControllerFact ruleControllerFact = new RuleControllerFact(marketingRule, dynamicKeyedBean.getEventBean(), triggerModelRuleMatchService, false,null);

            long s = System.currentTimeMillis();
            // 调用drools引擎，执行计算
            kieSession.insert(ruleControllerFact);
            kieSession.fireAllRules();
            // 获取计算结果
            boolean matchResult = ruleControllerFact.isMatchResult();
            long e = System.currentTimeMillis();
            rule_calc_timeAmount.inc(e-s);
            if(matchResult){
                rule_match_calc_timeAmount.inc(e-s);
                rule_is_match_calc_cntAmount.inc();

                ruleIsMatchAvgTimeGauge.incRuleIsMatchCount();
                ruleIsMatchAvgTimeGauge.incRuleIsMatchTimeAmount(e-s);

            }else{
                rule_not_match_calc_timeAmount.inc(e-s);
                rule_not_match_calc_cntAmount.inc();
            }


            //判断计算结果是否有定时条件
            if (matchResult){
                if (marketingRule.isOnTimer()) {
                    //这个规则是否是一个带定时条件的规则
                    //从规则中取出所有定时条件
                    List<TimerCondition> timerConditionList = marketingRule.getTimerConditionList();
                    for (TimerCondition timerCondition : timerConditionList) {
                        long triggerTime = dynamicKeyedBean.getEventBean().getTimeStamp() + timerCondition.getTimeLate();
                        //注册定时器
                        ctx.timerService().registerEventTimeTimer(triggerTime);
                        //在定时信息state进行记录
                        timeInfoState.add(Tuple2.of(ruleStateBean,triggerTime));
                    }
                } else {
                    //不带定时条件
                    out.collect(new RuleMatchResult(dynamicKeyedBean.getKeyValue(), marketingRule.getRuleId(),dynamicKeyedBean.getEventBean().getTimeStamp(),System.currentTimeMillis()));
                }
            }
        }
    }

    @Override
    public void processBroadcastElement(MaxwellLogBean maxwellLogBean, KeyedBroadcastProcessFunction<String, DynamicKeyedBean, MaxwellLogBean, RuleMatchResult>.Context context, Collector<RuleMatchResult> collector) throws Exception {

    }

    @Override
    public void onTimer(long timestamp, KeyedBroadcastProcessFunction<String, DynamicKeyedBean, MaxwellLogBean, RuleMatchResult>.OnTimerContext ctx, Collector<RuleMatchResult> out) throws Exception {
        Iterable<Tuple2<RuleStateBean, Long>> ruleTimerIterable = timeInfoState.get();
        Iterator<Tuple2<RuleStateBean, Long>> iterator = ruleTimerIterable.iterator();
        while(iterator.hasNext()){
            //规则及其注册定时器的触发时间
            Tuple2<RuleStateBean, Long> tp = iterator.next();
            //判断这个规则+定时点，是否对应本次触发点
            if (tp.f1 == timestamp) {
                RuleStateBean ruleStateBean = tp.f0;
                MarketingRule marketingRule = ruleStateBean.getMarketingRule();
                KieSession kieSession = ruleStateBean.getKieSession();
                //如果对应，检查该规则的定时条件(定时条件中包含的就是行为组合条件列表)
                for (TimerCondition timerCondition : marketingRule.getTimerConditionList()) {

                    //调用service去检查再条件指定的时间范围内，事件组合发生次数是否满足
                    TimerMatchDroolsFact timerMatchDroolsFact = new TimerMatchDroolsFact(ctx.getCurrentKey(), timerCondition, timestamp - timerCondition.getTimeLate(), timestamp, false);
                    RuleControllerFact ruleControllerFact = new RuleControllerFact();
                    ruleControllerFact.setTimerMatchDroolsFact(timerMatchDroolsFact);
                    kieSession.insert(ruleControllerFact);
                    kieSession.fireAllRules();
                    boolean b = timerMatchDroolsFact.isTimeMatchResult();
                    //清除已经检查完的规则定时点state信息
                    iterator.remove();
                    if (b){
                        RuleMatchResult ruleMatchResult = new RuleMatchResult(ctx.getCurrentKey(), marketingRule.getRuleId(), timestamp, ctx.timerService().currentProcessingTime());
                        out.collect(ruleMatchResult);
                    }
                }
            }
            //增加删除过期定时信息的逻辑(一般不会出现)
            if (tp.f1 < timestamp) {
                iterator.remove();
            }
        }

    }
}
